package com.neusoft.databus.agent.trap;

import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;

import com.neusoft.bizcore.common.bean.MessageBean;
import com.neusoft.bizcore.common.cache.CacheService;
import com.neusoft.bizcore.common.messaging.ProducerService;
import com.neusoft.bizcore.web.dto.result.ResultDTO;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.databus.agent.config.DatabusAgentProperties;
import com.neusoft.databus.agent.service.AssociateResourceService;
import com.neusoft.databus.common.bean.SnmptrapBean;
import com.neusoft.databus.common.constant.DatabusConstant;
import com.neusoft.databus.common.dto.ResourceDTO;

import lombok.extern.slf4j.Slf4j;

/**
 * SNMP告警处理
 *
 * @author sunchf
 * @date 2019年3月28日 上午8:52:49
 */
@Slf4j
@ConditionalOnProperty(name = "databus.agent.snmpTrap.server.disabled", havingValue = "false", matchIfMissing = true)
@Service
public class SnmptrapService implements SnmptrapProcessor {

    @Autowired
    private DatabusAgentProperties properties;
    @Autowired(required = false)
    private ProducerService producerService;
    @Autowired
    private AssociateResourceService associateResourceService;
    @Autowired
    private CacheService cacheService;
    private ArrayBlockingQueue<SnmptrapBean> queue = null;
    private final ExecutorService executorService = Executors.newSingleThreadExecutor();

    @PostConstruct
    public void init() {
        this.queue = new ArrayBlockingQueue<>(this.properties.getSnmpTrap().getServer().getBlockingQueueSize());
        this.executorService.execute(new ProcessTrapTask());
    }

    @Override
    public void process(final SnmptrapBean bean) {
        if (SnmptrapService.log.isDebugEnabled()) {
            SnmptrapService.log.debug("receive a snmp trap bean {}", bean);
        }
        // 开启消息中间件缓存，防止报警风暴丢失数据
        if (this.properties.getSnmpTrap().getServer().isL2CacheProvider()) {
            if (!this.queue.offer(bean)) {
                final String jsonStr = JsonUtils.pojoToJson(bean);
                final MessageBean messageBean = new MessageBean(jsonStr, new Date());
                if (null != this.producerService) {
                    this.producerService.send(this.properties.getSnmpTrap().getServer().getL2CacheTopic(), messageBean);
                }
            }
        } else {
            try {
                this.queue.put(bean);
            } catch (final InterruptedException e) {
                SnmptrapService.log.error(e.getMessage(), e);
            }
        }
    }

    // 起一个线程处理队列中的数据
    class ProcessTrapTask implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    final SnmptrapBean bean = SnmptrapService.this.queue.poll(3, TimeUnit.SECONDS);
                    if (null == bean) {
                        continue;
                    }
                    if (!SnmptrapService.this.setResourceCode(bean)) {
                        if (SnmptrapService.log.isWarnEnabled()) {
                            SnmptrapService.log.warn("associate relationship is not found in databus: {}", bean);
                        }
                        continue;
                    }
                    SnmptrapService.this.processTrap(bean);
                } catch (final InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public void processTrap(final SnmptrapBean bean) {
        if (SnmptrapService.log.isDebugEnabled()) {
            SnmptrapService.log.debug("process snmp trap from queue {}", bean);
        }
        final MessageBean messageBean = new MessageBean(JsonUtils.pojoToJson(bean), new Date());
        this.producerService.send(this.properties.getSnmpTrap().getServer().getToPolicyTopic(), messageBean);
    }

    /**
     * 根据子系统设备编码查询databus资源code
     *
     * @param bean
     * @return
     */
    private boolean setResourceCode(final SnmptrapBean bean) {
        final String cacheName = DatabusConstant.CACHE_NAME_ASSOCIATE;
        boolean result = false;
        try {
            final String associate = bean.getResourceId();
            if (StringUtils.isNotBlank(associate)) {
                final boolean isInCache = this.cacheService.isInCache(cacheName, associate);
                ResourceDTO model = null;
                if (isInCache) {
                    model = (ResourceDTO) this.cacheService.get(cacheName, associate);
                } else {
                    final ResultDTO<ResourceDTO> response =
                            SnmptrapService.this.associateResourceService.findResource(associate);
                    if (response.isSuccess() && (null != response.getData())) {
                        model = response.getData();
                        this.cacheService.put(cacheName, associate, model);
                    }
                }

                if (null != model) {
                    bean.setResourceId(model.getCode());
                    result = true;
                }
            }
        } catch (final Exception e) {
            //SnmptrapService.log.warn("{}", e.getMessage());
        }
        return result;
    }
}
